package com.apache.spark.app

import com.alibaba.fastjson.{JSON, JSONObject}
import com.apache.spark.utils.{MyKafkaUtils, MyOffsetsUtils, MyRedisUtils}
import com.twitter.chill.Input
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.TopicPartition
import org.apache.spark.SparkConf
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.{HasOffsetRanges, OffsetRange}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import redis.clients.jedis.Jedis

import java.util

/**
 * 业务数据分流消费
 *
 * 1.准备实时环境
 *
 * 2.从redis中读取偏移量
 *
 * 3.从Kafka中消费数据
 *
 * 4.提取偏移量结束点
 *
 * 5.数据处理
 * 5.1 转换数据结构
 * 5.2 分流
 * 事实数据 => Kafka
 * 维度数据 => Redis
 *
 * 6.flush Kafka的缓冲区
 *
 * 7.提交offset
 *
 *
 */
object OdsBaseDbApp {
  def main(args: Array[String]): Unit = {

    //1. 准备实时环境
    val conf: SparkConf = new SparkConf().setAppName("ods_base_db_app").setMaster("local[*]")
    val scc: StreamingContext = new StreamingContext(conf, Seconds(5))

    //拿到Kafka的消费者和消费者组
    val topicName: String = "ODS_BASE_DB"
    val groupId: String = "ODS_BASE_DB_GROUP"

    //2. 从redis中读取偏移量
    val offsets: Map[TopicPartition, Long] = MyOffsetsUtils.readOffset(topicName, groupId)

    //3. 从Kafka中消费数据
    var kafkaDStream: InputDStream[ConsumerRecord[String, String]] = null
    if (offsets != null && offsets.nonEmpty) {
      kafkaDStream = MyKafkaUtils.getKafkaDStream(scc, topicName, groupId, offsets)
    } else {
      kafkaDStream = MyKafkaUtils.getKafkaDStream(scc, topicName, groupId)
    }

    //4. 提取偏移量结束点
    var offsetRanges: Array[OffsetRange] = null
    val offsetRangesDStream: DStream[ConsumerRecord[String, String]] = kafkaDStream.transform(
      rdd => {
        offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
        rdd
      }
    )

    //5. 数据处理
    //5.1 转换数据结构
    val jsonObjDStream: DStream[JSONObject] = offsetRangesDStream.map(
      consumerRecord => {
        val dataJson: String = consumerRecord.value()
        val jSONObject: JSONObject = JSON.parseObject(dataJson)
        //返回给JSONObject
        jSONObject
      }
    )

    //测试 (成功)
    //jsonObjDStream.print(1000)

    //下一步，分流

    //    //事实表清单
    //    val factTables: Array[String] = Array[String]("order_info", "order_detail" /*缺啥补啥*/)
    //    //维度表清单
    //    val dimTables: Array[String] = Array[String]("user_info", "base_province" /*缺啥补啥*/)

    // Redis连接写在哪里？注意:连接对象它是不能进行传输的
    //foreachRDD 外面 : Driver 中 :  连接对象不能进行序列化，不能传输
    //foreachRDD 里面 ，foreachPartition外面 : Driver 中 : 连接对象不能进行序列化，不能传输
    //foreachPartition 里面 : Executor 中 : 每个分区数据开一个连接，用完关闭 (√)
    //foreachPartition 里面，循环里面 ：Executor : 每条数据开启一个连接，用完关闭，太频繁

    //5.2 分流
    jsonObjDStream.foreachRDD(
      rdd => {

        //思考 : 如何动态配置表清单 ??? (在不停掉程序，不修改配置的情况下)
        //将表清单维护到Redis中，实时任务中动态的到redis中获取表清单 -> 采用广播变量
        //类型: String,list,set,zset,hash  选择(set)
        //key: FACT:TABLES  DIM:TABLES
        //value: 表名的集合
        //写入API: sadd
        //读取API: smembers
        //过期: 不过期

        val redisFactKeys: String = "FACT:TABLES"
        val redisDimKeys: String = "DIM:TABLES"
        //获取redis连接
        val jedis: Jedis = MyRedisUtils.getJedisFromPool()
        //事实表清单
        val factTables: util.Set[String] = jedis.smembers(redisFactKeys)
        //广播变量事实表
        val factTablesBC: Broadcast[util.Set[String]] = scc.sparkContext.broadcast(factTables)
        //维度表清单
        val dimTables: util.Set[String] = jedis.smembers(redisDimKeys)
        //广播变量维度表
        val dimTablesBC: Broadcast[util.Set[String]] = scc.sparkContext.broadcast(dimTables)

        //打印测试
        println("factTables: " + factTables)
        println("dimTables: " + dimTables)

        //关闭连接
        jedis.close()

        rdd.foreachPartition(
          jsonObjIter => {
            //开启redis连接
            val jedis: Jedis = MyRedisUtils.getJedisFromPool()
            for (jsonObj <- jsonObjIter) {
              //提取操作数据类型
              val operType: String = jsonObj.getString("type")
              //对操作类型，进行模式匹配
              val opValue: String = operType match {
                // 加入Maxwell 全量导入，
                case "bootstrap-insert" => "I"
                case "insert" => "I"
                case "update" => "U"
                case "delete" => "D"
                case _ => null // create select
              }

              //判断操作类型 1. 明确什么操作   2. 过滤不感兴趣的数据
              if (opValue != null) {
                //提取表名
                val tableName: String = jsonObj.getString("table")
                //提取数据
                val data: String = jsonObj.getString("data")

                //判断是否为事实表
                if (factTablesBC.value.contains(tableName)) {
                  //事实数据
                  //例如 DWD_ORDER_INFO_I ,DWD_ORDER_INFO_U ,DWD_ORDER_INFO_D
                  val dwdTopicName: String = s"DWD_${tableName.toUpperCase}_${opValue}"
                  //发送
                  MyKafkaUtils.send(dwdTopicName, data)
                }

                //模拟数据延迟
                if (tableName.equals("order_detail")){
                  Thread.sleep(200)
                }

                if (dimTablesBC.value.contains(tableName)) {
                  // 维度数据
                  // 类型：String list set zset hash
                  // hash : 整个表存成一个hash,要考虑数据量大小和将来数据量增长问题及高频访问问题 一条数据存成一个hash
                  // String : 一条数据存成一个 jsonString
                  // key : DIM:表名:ID
                  // value : 整条数据的jsonString
                  // 写入API : set
                  // 读取API : get
                  // 过期 : 不过期


                  //提取数据中的id
                  val dataObj: JSONObject = jsonObj.getJSONObject("data")
                  val id: String = dataObj.getString("id")
                  //key
                  val redisKey: String = s"DIM:${tableName.toUpperCase}:$id"
                  //redis工具类,写入redis(String类型)
                  //在此处开Redis连接的时候太频繁了
                  //val jedis: Jedis = MyRedisUtils.getJedisFromPool()
                  jedis.set(redisKey, dataObj.toJSONString)
                  //在xshell中 ，如果要看redis中文，命令是 redis-cli --raw
                  //关闭redis
                  //jedis.close()
                }
              }
            }

            //关闭redis连接
            jedis.close()

            //刷写 Kafka 缓冲区
            MyKafkaUtils.flush()
          }
        )
        //提交offset
        MyOffsetsUtils.saveOffset(topicName, groupId, offsetRanges)
      }
    )

    scc.start()
    scc.awaitTermination()
  }
}
